package com.hxq.demo1;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

/**按时间消费
 * 需求：在生产环境中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。
 * 例如要求按照时间消费前一天的数据，怎么处理？
 * @author hxq
 * @date 2022/4/22 16:23
 */
public class SeekTimeConsumer {

    public static void main(String[] args) {
        autoCommitOffset();
    }

    private static void autoCommitOffset(){

        Properties properties = setConfigInfo(true);
        //创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        //订阅主题
        consumer.subscribe(Arrays.asList("TestTopic"));

        Set<TopicPartition> assignment= new HashSet<>();//获取分配信息

        while (assignment.size() == 0) {//程序执行很快，消费者组可能还没有初始化好
            consumer.poll(Duration.ofSeconds(1));
        // 获取消费者分区分配信息（有了分区分配信息才能开始消费）
            assignment = consumer.assignment();
        }

        Map<TopicPartition, Long> timestampToSearch = new HashMap<>();
        // 封装集合存储，每个分区对应一天前的数据
        for (TopicPartition topicPartition : assignment) {
            timestampToSearch.put(topicPartition, System.currentTimeMillis() - 3 * 24 * 3600 * 1000);
        }
        // 获取从 1 天前开始消费的每个分区的 offset
        Map<TopicPartition, OffsetAndTimestamp> offsets = consumer.offsetsForTimes(timestampToSearch);

        // 遍历每个分区，对每个分区设置消费时间。
        for (TopicPartition tp: assignment) {
            OffsetAndTimestamp offsetAndTimestamp = offsets.get(tp);
            // 根据时间指定开始消费的位置
            if (offsetAndTimestamp != null){
                consumer.seek(tp, offsetAndTimestamp.offset());
            }
        }

        while (true){
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records){
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }

        }

    }



    private static Properties setConfigInfo(Boolean isAuto){
        Properties props = new Properties();
        props.put("bootstrap.servers", "212.129.155.35:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", isAuto);//是否开启自动提交 offset 功能
        props.put("auto.commit.interval.ms", "1000");//自动提交 offset 的时间间隔
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());

        return props;
    }
}
